package main

import (
	"context"
	"encoding/json"
	"github.com/streadway/amqp"
	"go.uber.org/zap"
	"mts/log_srv/internal/repository"
	"mts/log_srv/internal/repository/dao"
	"mts/log_srv/internal/service"
	"mts/log_srv/internal/web"
	"mts/log_srv/ioc"
	"sync"
)

var logger *zap.SugaredLogger

func init() {
	ioc.InitLogger()
	logger = zap.S().With("module", "main")
}

func failOnError(err error, msg string) {
	if err != nil {
		zap.S().Fatalf("%s: %s", msg, err)
	}
}

func main() {
	ioc.InitConfig()
	db := ioc.InitDB()
	//dao
	logsDao := dao.NewLogsDao(db)
	langDao := dao.NewLangDao(db)
	//repo
	logRepo := repository.NewLogRepo(logsDao)
	langRepo := repository.NewLangRepo(langDao)
	//svc
	logService := service.NewLogService(logRepo)
	langService := service.NewLangService(langRepo)
	//web
	logHandler := web.NewLogHandler(logService)
	langHandler := web.NewLangLogHandler(langService)

	// 初始化 RabbitMQ
	conn, ch, err := ioc.InitRabbitMQ()
	failOnError(err, "RabbitMQ 初始化失败")
	defer conn.Close()
	defer ch.Close()

	// 声明主队列 + 死信配置
	q := declareMainQueue(ch)
	// 声明死信交换机与死信队列
	declareDeadLetterComponents(ch)

	// 启动消费者
	var wg sync.WaitGroup
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	// 启动 RabbitMQ 消费者协程
	startLogConsumer(&wg, ctx, ch, q.Name, logService.HandleLog)
	startDLQConsumer(&wg, ctx, ch, logService.HandleLog)

	// 启动 Web 服务
	wg.Add(1)
	go func() {
		defer wg.Done()
		srv := ioc.InitWebServer(logHandler, langHandler) // 根据实际参数传入 handler
		err := srv.Run(":8033")
		if err != nil {
			logger.Fatalf("服务启动失败: %s", err.Error())
		}
	}()

	wg.Wait()
}

// 声明主队列并配置死信参数
func declareMainQueue(ch *amqp.Channel) amqp.Queue {
	args := amqp.Table{
		"x-dead-letter-exchange":    "dlx_logs_exchange",
		"x-dead-letter-routing-key": "dlq.logs.key",
		"x-message-ttl":             60000,       //消息超时时间
		"x-overflow":                "drop-head", //这里选择丢弃头部旧消息而不是直接拒绝新消息。
	}

	q, err := ch.QueueDeclare(
		"logs_queue",
		true,
		false,
		false,
		false,
		args,
	)
	failOnError(err, "Failed to declare logs_queue")
	return q
}

// 声明死信交换机和队列，并绑定
func declareDeadLetterComponents(ch *amqp.Channel) {
	if err := ch.ExchangeDeclare("dlx_logs_exchange", "direct", true, false, false, false, nil); err != nil {
		zap.S().Fatalf("Failed to declare DLX: %v", err)
	}
	//防止死信队列满,声明死信队列时添加超时机制,超时丢弃队列中的消息
	if _, err := ch.QueueDeclare("dlq_logs_queue", true, false, false, false, amqp.Table{
		"x-message-ttl": 7200000, //死信队列消息最多存活3h
	}); err != nil {
		zap.S().Fatalf("Failed to declare DLQ: %v", err)
	}
	if err := ch.QueueBind("dlq_logs_queue", "dlq.logs.key", "dlx_logs_exchange", false, nil); err != nil {
		zap.S().Fatalf("Failed to bind DLQ: %v", err)
	}
}

// 启动主队列消费者
func startLogConsumer(wg *sync.WaitGroup, ctx context.Context, ch *amqp.Channel, queueName string, handler func(map[string]interface{}) error) {
	msgs, err := ch.Consume(queueName, "", true, false, false, false, nil)
	failOnError(err, "Failed to register consumer")
	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case d, ok := <-msgs:
				if !ok {
					return
				}
				handleMessage(d, handler, "Log received")
			case <-ctx.Done():
				return
			}
		}

	}()
}

// 启动死信队列消费者
func startDLQConsumer(wg *sync.WaitGroup, ctx context.Context, ch *amqp.Channel, handler func(map[string]interface{}) error) {
	dlqMsgs, err := ch.Consume("dlq_logs_queue", "", false, false, false, false, nil)
	failOnError(err, "Failed to register DLQ consumer")

	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case d, ok := <-dlqMsgs:
				if !ok {
					return
				}
				if err := handleMessageWithAck(d, handler, "Dead letter message received", ch); err != nil {
					_ = ch.Nack(d.DeliveryTag, false, false)
				} else {
					_ = ch.Ack(d.DeliveryTag, false)
				}
			case <-ctx.Done():
				return
			}
		}
	}()
}

// 封装通用消息处理逻辑
func handleMessage(d amqp.Delivery, handler func(map[string]interface{}) error, logPrefix string) {
	var data map[string]interface{}
	if err := json.Unmarshal(d.Body, &data); err != nil {
		zap.S().Errorf("Error decoding JSON: %v", err)
		return
	}
	zap.S().Infof("%s: %v", logPrefix, data)
	if err := handler(data); err != nil {
		zap.S().Errorf("Error handling message: %v", err)
	}
}

// 带 Ack/Nack 处理的消息封装
func handleMessageWithAck(d amqp.Delivery, handler func(map[string]interface{}) error, logPrefix string, ch *amqp.Channel) error {
	var data map[string]interface{}
	if err := json.Unmarshal(d.Body, &data); err != nil {
		logger.Errorf("Error decoding DLQ JSON: %v", err)
		return err
	}
	zap.S().Warnf("%s: %v", logPrefix, data)
	return handler(data)
}
